package com.atguigu.realtime.util;

import com.atguigu.realtime.common.Constant;
import com.sun.istack.Nullable;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaUtil {

    static String BOOTSTRAP_SERVERS = "node02:6667";
    static String DEFAULT_TOPIC = "default_topic";



//        Properties props = new Properties();
//        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
//        props.setProperty("isolation.level","read_committed");
//        props.setProperty("group.id", "groupId");
//        return new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),props);
        public static FlinkKafkaConsumer<String>  getKafkaConsumer(String topic, String groupId) {
            Properties prop = new Properties();
            prop.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
            prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

            FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(topic,
                    new KafkaDeserializationSchema<String>() {
                        @Override
                        public boolean isEndOfStream(String nextElement) {
                            return false;
                        }

                        @Override
                        public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                            if(record != null && record.value() != null) {
                                return new String(record.value());
                            }
                            return null;
                        }

                        @Override
                        public TypeInformation<String> getProducedType() {
                            return TypeInformation.of(String.class);
                        }
                    }, prop);
            return consumer;
        }

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {

//        Properties prop = new Properties();
//        prop.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
//        prop.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60 * 15 * 1000 + "");
//
//        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<String>(DEFAULT_TOPIC, new KafkaSerializationSchema<String>() {
//
//            @Override
//            public ProducerRecord<byte[], byte[]> serialize(String jsonStr, @Nullable Long timestamp) {
//                return new ProducerRecord<byte[], byte[]>(topic, jsonStr.getBytes());
//            }
//        }, prop, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
//
//        return producer;

        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(
                "node02:6667",  // Kafka地址
                topic, // 要写入的kakfa的topic
                new SimpleStringSchema()  // 序列化器 （这里是序列化过程）  继承了 DeserializationSchema<> 反序列化, SerializationSchema<>序列化
        );
        return producer;
    }

    /**
     * Kafka-Source DDL 语句
     *
     * @param topic   数据源主题
     * @param groupId 消费者组
     * @return 拼接好的 Kafka 数据源 DDL 语句
     */
    public static String getKafkaDDL(String topic, String groupId) {

        return " with ('connector' = 'kafka', " +
                " 'topic' = '" + topic + "'," +
                " 'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVERS + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'format' = 'json', " +
                " 'scan.startup.mode' = 'group-offsets')";
    }

    /**
     * Kafka-Sink DDL 语句
     *
     * @param topic 输出到 Kafka 的目标主题
     * @return 拼接好的 Kafka-Sink DDL 语句
     */
    public static String getKafkaSinkDDL(String topic) {
        return "WITH ( " +
                "  'connector' = 'kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVERS + "', " +
                "  'format' = 'json' " +
                ")";
    }

    /**
     * UpsertKafka-Sink DDL 语句
     *
     * @param topic 输出到 Kafka 的目标主题
     * @return 拼接好的 UpsertKafka-Sink DDL 语句
     */
    public static String getUpsertKafkaDDL(String topic) {

        return "WITH ( " +
                "  'connector' = 'upsert-kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVERS + "', " +
                "  'key.format' = 'json', " +
                "  'value.format' = 'json' " +
                ")";
    }


}
